use std::{
    collections::HashMap,
    env::{self},
    time::Duration,
};

use anyhow::bail;
use apollo_client::conf::{
    meta::IpValue,
    requests::{FetchRequest, WatchRequest},
    ApolloConfClientBuilder,
};
use cidr_utils::cidr::IpCidr;

use dotenv::dotenv;
use futures_util::{pin_mut, StreamExt};
use r2d2_redis::{
    r2d2::{self, Pool},
    RedisConnectionManager,
};
use sea_orm::{ConnectOptions, Database};

use crate::system::{app::APOLLO_CONF, env_util, error::ErrorCode};
use tracing::{info, Level};
use tracing_appender::rolling;
use url::Url;

use super::{error::AppError, result::Result};

pub(crate) async fn async_init() -> Result<bool, anyhow::Error> {
    let r = apollo_init().await;
    info!("the r {:?}", r);
    if r.is_err() {
        bail!(AppError::BizError {
            code: ErrorCode::SYSTEM_ERROR_PARSE.get_code(),
            desc: r.err().unwrap().to_string(),
        });
    }
    Ok(true)
}

pub(crate) fn sync() {
    env_init();
    logger_init();
}

pub(crate) async fn watch() {
    tokio::spawn(async {
        let _ = apollo_watch().await;
    });
}

fn env_init() -> bool {
    let ret = dotenv().ok();
    ret.is_some()
}

async fn apollo_init() -> Result<(), Box<dyn std::error::Error>> {
    let apollo_address_ret = env_util::get("APOLLO_ADDRESS");
    if apollo_address_ret.is_err() {
        return Err(apollo_address_ret.err().unwrap().to_box());
    }
    let client = ApolloConfClientBuilder::new_via_config_service(Url::parse(
        apollo_address_ret.unwrap().as_str(),
    )?)?
    .build()?;

    let apollo_namespaces_ret = env_util::get("APOLLO_NAMESPACES");
    if apollo_namespaces_ret.is_err() {
        return Err(apollo_namespaces_ret.err().unwrap().to_box());
    }
    for i in apollo_namespaces_ret.unwrap().split(",") {
        let config = client
            .fetch(FetchRequest {
                app_id: env::var("APOLLO_APP_ID").unwrap().to_string(),
                namespace_name: i.to_string(),
                ip: Some(IpValue::HostName),
                ..Default::default()
            })
            .await?;
        let config = config.configurations;
        info!("apollo.application : {:?}", config);
        let mut apollo_app_guard = APOLLO_CONF.try_write().unwrap();
        *apollo_app_guard = config;
        tracing::info!("apollo init on application success.");
    }
    Ok(())
}

async fn apollo_watch() -> Result<(), Box<dyn std::error::Error>> {
    //
    let apollo_address_ret = env_util::get("APOLLO_ADDRESS");
    if apollo_address_ret.is_err() {
        return Err(apollo_address_ret.err().unwrap().to_box());
    }
    let client = ApolloConfClientBuilder::new_via_config_service(Url::parse(
        apollo_address_ret.unwrap().as_str(),
    )?)?
    .build()?;

    let apollo_namespaces_ret = env_util::get("APOLLO_NAMESPACES");
    if apollo_namespaces_ret.is_err() {
        return Err(apollo_namespaces_ret.err().unwrap().to_box());
    }
    let mut apollo_namespaces: Vec<String> = Vec::new();
    for i in apollo_namespaces_ret.unwrap().split(",") {
        apollo_namespaces.push(i.to_string());
    }
    let stream = client.watch(WatchRequest {
        app_id: env::var("APOLLO_APP_ID").unwrap().to_string(),
        namespace_names: apollo_namespaces,
        ip: Some(IpValue::HostCidr(IpCidr::from_str(
            env::var("APOLLO_IP_CIDER").unwrap().as_str(),
        )?)),
        ..Default::default()
    });
    pin_mut!(stream);
    let mut counter = 1;
    tracing::info!("apollo watch on application success.");
    while let Some(response) = stream.next().await {
        let res = response?;
        // jump out first
        if counter == 1 {
            counter += 1;
            continue;
        }
        for r in res {
            info!("apollo watch event : {:?}", r);
            let new_config = r.1.unwrap().configurations;
            let mut apollo_app_guard = APOLLO_CONF.try_write().unwrap();
            *apollo_app_guard = new_config;
        }
    }
    Ok(())
}

fn logger_init() -> bool {
    let logger_file = rolling::daily("./logs", "zino");
    tracing_subscriber::fmt()
        .with_writer(logger_file)
        .with_ansi(false)
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| {
                "example_tracing_aka_logging=debug,tower_http=debug,axum::rejection=trace".into()
            }),
        )
        .with_max_level(Level::TRACE)
        .json()
        .init();
    tracing::info!("logger init on application success.");
    true
}
/**
 *
 */
pub(crate) fn redis_init() -> HashMap<String, Pool<RedisConnectionManager>> {
    let app = APOLLO_CONF.try_read().unwrap();
    // main redis build connection
    let url = format!(
        "redis://{}:{}",
        app.get("redis.main.host").unwrap(),
        app.get("redis.main.port").unwrap()
    );
    let manager = RedisConnectionManager::new(url.clone()).unwrap();
    let pool = r2d2::Pool::builder()
        .min_idle(Some(5))
        .max_size(150)
        .max_lifetime(Some(Duration::from_secs(3600)))
        .build(manager);
    // log
    tracing::info!("redis init on {} success.", url);
    let mut hash = HashMap::new();
    hash.insert(String::from("main"), pool.unwrap());
    //
    hash
}
/**
 *
 * todo 有时候，这个Database::connect会一直阻塞，无返回
 * date 2023-10-26
 */
pub(crate) fn db_init() -> HashMap<String, sea_orm::DatabaseConnection> {
    let app = APOLLO_CONF.try_read().unwrap();

    let db_main_url = format!(
        "mysql://{}:{}@{}:{}/{}",
        app.get("mysql.main.user").unwrap(),
        match app.get("mysql.main.password") {
            Some(v) => v,
            None => "",
        },
        app.get("mysql.main.host").unwrap(),
        app.get("mysql.main.port").unwrap(),
        app.get("mysql.main.db").unwrap()
    );
    let mut main_pool_opt = ConnectOptions::new(db_main_url.clone());
    main_pool_opt
        .max_connections(150)
        .min_connections(5)
        .connect_timeout(Duration::from_secs(3))
        .acquire_timeout(Duration::from_secs(3))
        .acquire_timeout(Duration::from_secs(3))
        .idle_timeout(Duration::from_secs(300))
        .max_lifetime(Duration::from_secs(3000))
        .sqlx_logging(true);

    tracing::info!("mysql init on {} starting.", db_main_url.clone());
    let main_conn =
        futures::executor::block_on(async { Database::connect(main_pool_opt).await.unwrap() });
    tracing::info!("mysql init on {} success.", db_main_url.clone());
    let mut hash = HashMap::new();
    hash.insert(String::from("main"), main_conn);
    hash
}
